#include "ReactorSession.h"
#include "ThreadPool.h"
ReactorSession::ReactorSession(int _idx)
{
	m_index = _idx;
	Run();
}

ReactorSession::~ReactorSession()
{
}

void ReactorSession::Run()
{
	std::thread(std::bind(&ReactorSession::NetIOEvent, this)).detach();
}

void ReactorSession::AddNewSession(const SOCKET& _fd)
{
	std::lock_guard<std::mutex> lock(m_mutex);
	auto& session = m_clients[_fd];
	session.m_fd = _fd;
}

void ReactorSession::OnReadNetData(const SOCKET& _fd)
{
	TimeStepTmp::GetInstance()->m_recvCount++;
	std::unique_lock<std::mutex> lock(m_mutex);
	auto it = m_clients.find(_fd);
	if (it != m_clients.end())
	{
		it->second.m_busy = true;
		auto res = pool->Exec(std::bind(&ClientSession::OnReadNetData, &it->second));
		lock.unlock();
		res.get();
	}
}

void ReactorSession::NetIOEvent()
{
	fd_set m_readFds;
	fd_set m_writeFds;
	fd_set m_errFds;
	int32_t m_maxFd = 0;
	timeval timeout = { 0,10 };
	while (true)
	{
		//NotTimeMsg();
		memset(&m_readFds, 0, sizeof(m_readFds));
		memset(&m_writeFds, 0, sizeof(m_writeFds));
		memset(&m_errFds, 0, sizeof(m_errFds));
		auto fds = GetAllFds();
		if (fds.empty())
		{
			std::this_thread::sleep_for(std::chrono::microseconds(50));
			continue;
		}
		for (const auto& fd : fds)
		{
			FD_SET(fd, &m_readFds);
			FD_SET(fd, &m_writeFds);
			FD_SET(fd, &m_errFds);
			m_maxFd = max(fd, m_maxFd);
		}

		int32_t result = select(m_maxFd + 1, &m_readFds, &m_writeFds, &m_errFds, &timeout);
		if (SOCKET_ERROR == result)
		{
			return;
		}
		for (const auto& fd : fds)
		{
			if (FD_ISSET(fd, &m_readFds))
			{
				OnReadNetData(fd);
			}
		}
	}
}
